package demo.designPattern.observer;

import demo.java.lang.ThreadDemo;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.time.Duration;
import java.util.concurrent.CompletableFuture;

/**
 * Project Reactor 是一个基于 Reactive Streams 规范的反应式编程框架，旨在提供高效的异步处理能力，并能够以一种更简单和更流畅的方式处理数据流。
 * <p>
 * 关键组件
 * reactor-core 包主要包含以下几个核心组件：
 *
 * <li>Flux：这是一个表示异步序列的类，可以发出零个或多个数据项，以及完成或错误的信号。Flux 支持多项输出，适合处理多个异步事件。
 * <li>Mono：这是一个表示只有一个或没有数据项的异步序列的类。它类似于 Optional 和 Future 的概念，但以非阻塞的方式操作。
 * <li>Schedulers：提供用于执行操作的调度器，可以在特定的线程中运行操作，支持并发和并行执行。
 * <li>Operators：reactor-core 提供了一系列操作符（如 map、flatMap、filter、merge、zip 等），可以对 Flux 和 Mono 进行丰富的数据转换和处理。
 * <li>Backpressure：支持背压机制，以便消费者可以控制生成者数据生成的速度，确保系统的稳定。
 * <li>Exception Handling：提供错误处理机制，可以灵活地处理链中的错误，如 onErrorResume 和 retry 等操作。
 * <p>
 * 应用场景
 * <li>Web 应用：在反应式 Web 框架（如 Spring WebFlux）中使用，能够处理高并发的 HTTP 请求。
 * <li>微服务：在微服务架构中实现非阻塞的服务通信。
 * <li>数据流处理：处理实时数据流、事件驱动架构（EDA）、或高频交易等场景。
 * <li>数据访问：与数据库交互时，使用反应式驱动的 DAO。
 */
public class ReactorDemo {

    private static final Logger logger = LoggerFactory.getLogger(ReactorDemo.class);

    /**
     * 用于表示一个异步的、可以发出零个或多个项的序列。Flux 在反应式编程中扮演着至关重要的角色，适合处理多项输出的场景，如实时数据流、事件处理等。
     * 基本特性
     * <li>多项支持：Flux 可以发出任意数量的元素，包括零个或多个。这意味着它适用于处理诸如事件流或批量数据等场景。
     * <li>异步与非阻塞：Flux 是异步的且非阻塞的。这使得它可以在高负载情况下处理数据，同时确保应用程序的响应能力。
     * <li>背压：Flux 支持背压机制，允许消费者控制生产者数据生成的速度。背压是一种防止消费者被高流量数据淹没的机制。
     * <li>热/冷序列：
     * <li>- 冷序列：只有在消费者订阅时，生产者才开始生成数据。
     * <li>- 热序列：生产者在任何订阅者到来之前就开始生成数据。
     */
    @Test
    public void testFlux() {
        // 创建一个 Flux
        Flux<String> flux = Flux.just("A", "B", "C")
                .map(String::toLowerCase) // 转换为小写
                .filter(s -> s.startsWith("a")); // 过滤以 "a" 开头的项
        // 生成 1 到 5
        Flux<Integer> rangeFlux = Flux.range(1, 5);
        // 每秒产生一个数字
        Flux<Long> intervalFlux = Flux.interval(Duration.ofSeconds(1)).take(5);

        Flux<Integer> numberFlux = Flux.range(1, 10)
                .delayElements(Duration.ofMillis(500)) // 每个元素延迟500毫秒
                .filter(n -> n % 2 == 0);// 只保留偶数

        // 订阅 Flux
        numberFlux.subscribe(
                item -> logger.info("Received: {}", item), // onNext
                error -> logger.error("Error: ", error),   // onError
                () -> logger.info("[done]")      // onComplete

        );

        ThreadDemo.safeSleep(10_000);
    }

    // 验证数据
    private static Mono<String> validateData(String data) {
        return Mono.fromCallable(() -> {
            if (data == null || data.isEmpty()) {
                throw new IllegalArgumentException("Data is invalid");
            }
            System.out.println("验证数据: " + data);
            return data;
        });
    }

    // 处理数据
    private static Mono<String> processData(String data) {
        return Mono.fromCallable(() -> {
            String processedData = data.toUpperCase(); // 模拟数据处理
            System.out.println("处理数据: " + processedData);
            return processedData;
        });
    }

    // 保存数据
    private static Mono<String> saveData(String data) {
        return Mono.fromRunnable(() -> {
            // 模拟保存数据到数据库
            System.out.println("保存数据: " + data);
        }).thenReturn("Data saved successfully"); // 返回保存成功消息
    }

    @Test
    public void testDoOnSubscribe() {
        System.out.println("-----------------------------------doOnSubscribe");
        Flux.just("A", "B", "C").delayElements(Duration.ofSeconds(1))
                .doOnSubscribe(subscription -> logger.info("订阅开始"))
                .doFinally(signalType -> logger.info("订阅结束"))
                .subscribe(logger::info);


        ThreadDemo.safeSleep(10_000);
    }

    @Test
    public void testDoOnNext() {

        System.out.println("-----------------------------------doOnNext");
        Flux.just("A", "B", "C")
                .delayElements(Duration.ofSeconds(1))
                .doOnSubscribe(subscription -> logger.info("订阅开始: doOnSubscribe"))
                .doOnNext(subscription -> logger.info("订阅开始:doOnNext"))
                .doFinally(signalType -> logger.info("订阅结束:doFinally"))
                .subscribe(logger::info);

        ThreadDemo.safeSleep(10_000);
    }

    /**
     * Mono 是 Project Reactor 中的一个核心类，用于表示 异步的单值数据流。
     * 它是 Reactive Streams 的 Publisher 实现之一，专门用于处理最多包含一个元素的数据流。
     * Mono 非常适合表示可能为空或只包含一个结果的异步操作，例如 HTTP 请求、数据库查询或单个事件。
     */
    @Test
    public void testMono() {
        // 创建一个 Mono
        Mono<String> mono = Mono.just("Hello Mono!");
        // map: 对 Mono 中的值进行转换
        Mono<Integer> mappedMono = mono.map(String::length);
        // flatMap: 对 Mono 中的值进行转换，并将其转换为另一个Mono
        Mono<Integer> flatMappedMono = mono.flatMap(s -> Mono.just(s.length()));
        //        doOnNext
        mono.doOnNext(System.out::println).subscribe();
        // zip: 将多个 Mono 合并为一个 Mono
        Mono<String> zippedMono = Mono.zip(mono, flatMappedMono, (s1, s2) -> s1 + " " + s2);
        zippedMono.subscribe(System.out::println);
        // 忽略 Mono 的值，返回另一个 Mono
        Mono<Void> thenMono = mono.then(Mono.empty());
        // onErrorResume: 当 Mono 发出错误时，返回一个新的 Mono
        Mono<String> errorMono = Mono.error(new RuntimeException("Error"));
        Mono<String> fallbackMono = errorMono.onErrorResume(e -> Mono.just("Fallback"));

        // 订阅并处理值
        mappedMono.subscribe(
                value -> System.out.println("Received: " + value), // onNext
                error -> System.err.println("Error: " + error),    // onError
                () -> System.out.println("Completed")              // onComplete
        );

        // 订阅 Mono
        mono.subscribe(System.out::println);

        // 与 CompletableFuture 的互操作
        // 创建一个 CompletableFuture
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello");

        // 将 CompletableFuture 转换为 Mono
        Mono<String> futureMono = Mono.fromFuture(future);

        // 订阅并处理值
        futureMono.subscribe(System.out::println);
    }

    /**
     * then() 是 Mono 和 Flux 中的一个重要操作符，用于在 当前数据流完成后 执行另一个操作。
     * 它的核心特点是 忽略当前数据流的值，只关注其完成信号，并在完成后触发下一个操作。
     * then() 通常用于链式操作中，特别是当你只关心操作的完成状态而不需要处理具体数据时。
     */
    @Test
    public void testThen() {
        Mono.just("then()基本用法：忽略值，只关注完成信号")
                .doOnNext(System.out::println) // 打印 "Hello"
                .then() // 忽略值，只关注完成信号
                .subscribe(unused -> {
                    System.err.println("Completed"); // 打印 "Completed"
                });
        // 使用 then() 触发另一个 Mono
        Mono.just("使用 then() 触发另一个 Mono")
                .doOnNext(System.out::println) // 打印 "Hello"
                .then(Mono.just("World")) // 触发另一个 Mono
                .subscribe(System.out::println); // 打印 "World"
        //        触发另一个 Publisher：then(Publisher<V>)
        Mono.just("触发另一个 Publisher：thenMany(Publisher<V>)")
                .doOnNext(System.out::println) // 打印 "Hello"
                .thenMany(Flux.just("World", "!")) // 触发另一个 Publisher
                .subscribe(System.out::println); // 打印 "World" 和 "!"
        // 忽略值并触发另一个 Mono<Void>：thenEmpty(Mono<Void>)
        Mono.just("忽略值并触发另一个 Mono<Void>：thenEmpty(Mono<Void>)")
                .doOnNext(System.out::println) // 打印 "Hello"
                .thenEmpty(Mono.fromRunnable(() -> System.out.println("Done"))) // 触发另一个 Mono<Void>
                .subscribe();
        // 顺序操作
        Mono<Void> saveToDatabase = Mono.fromRunnable(() -> logger.info("保存数据到数据库"));
        Mono<Void> sendNotification = Mono.fromRunnable(() -> logger.info("然后发送通知"));
        saveToDatabase.then(sendNotification) // 先保存数据，然后发送通知
                .subscribe();
        // 忽略中间结果
        Mono.just("Step 1")
                .doOnNext(System.out::println)
                .then(Mono.just("Step 2"))
                .doOnNext(System.out::println)
                .then(Mono.just("Step 3"))
                .subscribe(a -> logger.info("收到: {}", a));
    }

    /**
     * doFinally()是Reactor中的一个重要操作符，它允许我们在信号流结束时（无论是正常完成、错误还是取消）执行一些操作。主要特点
     * <li>保证执行：无论流是如何结束的（完成、错误或取消），doFinally都会被执行
     * <li>执行时机：在所有其他终端操作（如onComplete、onError）之后执行
     * <li>可以获知结束原因：通过SignalType参数可以知道流是如何结束的
     * <li>线程安全：doFinally的操作保证在适当的线程上执行
     * <p><br/>
     * 注意事项
     * <li>不要在doFinally中抛出异常，因为这些异常不会被下游处理
     * <li>doFinally应该用于执行简单的清理操作，不应包含复杂的业务逻辑
     * <li>如果需要在清理操作完成后执行其他操作，应考虑使用using或usingWhen操作符
     * <li>doFinally不会改变流中的数据，它只是一个副作用操作符
     */
    @Test
    public void testDoFinally() {
        System.out.println("------------------正常完成:ON_COMPLETE");
        Mono.just("Hello")
                .doOnNext(System.out::println) // 打印 "Hello"
                .doFinally(signalType -> System.out.println("结束类型: " + signalType)) // ON_COMPLETE
                .subscribe();
        System.out.println("------------------发生错误:ON_ERROR");
        Mono.error(new RuntimeException("异常"))
                .doFinally(signalType -> {
                    System.out.println("结束类型: " + signalType);
                }).subscribe(
                        value -> System.out.println(value),
                        error -> System.err.println("Error: " + error)
                );
        System.out.println("------------------取消订阅:ON_CANCEL");
        Mono<String> mono = Mono.just("Hello")
                .delayElement(Duration.ofSeconds(2))
                .doFinally(signalType -> System.out.println("结束类型: " + signalType));

        Disposable disposable = mono.subscribe();
        disposable.dispose(); // 取消订阅
    }

    /**
     * using 和 usingWhen 是 Reactor 中用于资源管理的高级操作符，它们提供了一种安全和优雅的方式来处理需要获取和释放的资源。
     * using(): 专门为资源管理设计，提供了完整的资源获取、使用和释放的生命周期管理
     * 参数说明：
     * <li>resourceSupplier: 提供资源的函数
     * <li>resourceClosure: 使用资源进行操作的函数
     * <li>resourceCleanup: 清理资源的函数
     */
    @Test
    public void testUsing() {
        Mono<String> mono = Mono.using(
                Resource::new,                    // 创建资源
                resource -> Mono.just(resource.getData()),// 使用资源
                Resource::close                          // 清理资源
        );

        mono.subscribe(System.out::println);
    }

    /**
     * 链式操作
     */
    @Test
    public void testChainOperations() {
        // 模拟输入数据
        String inputData = "Valid Data";
        // 链式操作
        Mono.just(inputData)
                .flatMap(data -> validateData(data)) // 验证数据
                .flatMap(validData -> processData(validData)) // 处理数据
                .flatMap(processedData -> saveData(processedData)) // 保存数据
                .subscribe(
                        result -> System.out.println("Success: " + result), // 成功处理
                        error -> System.err.println("Error: " + error), // 错误处理
                        () -> System.out.println("Completed") // 完成信号
                );
    }

    /**
     * merge 操作符将多个数据流中的项进行合并，产生一个新的 Flux，这个新的 Flux 会发出所有源流中的元素，而不关心它们的顺序。
     * 当多个流的元素同时到达时，元素会立即被发出，而不需要等待其他流的元素。
     * merge 输出的数量是所有输入流中元素数量的总和。
     */
    @Test
    public void testMerge() {
        Flux<Integer> flux1 = Flux.just(1, 2, 3);
        Flux<Integer> flux2 = Flux.just(4, 5, 6);

        Flux<Integer> mergedFlux = Flux.merge(flux1, flux2);

        mergedFlux.subscribe(m -> logger.info("received: {}", m)); // 输出可能是 1, 2, 3, 4, 5, 6 以不确定顺序
    }

    /**
     * map 和 flatMap 的区别
     * <li>特性: 	map	VS flatMap
     * <li>转换类型：	同步转换 VS	异步转换
     * <li>返回值:	返回转换后的值 VS	返回一个 Publisher（如 Mono 或 Flux）
     * <li>适用场景	简单的值转换	VS 复杂的异步操作（如调用外部服务、数据库查询等）
     * <li>数据流处理	一对一映射 VS	一对多映射（如果返回的是 Flux）
     * <li>性能	更高效，适合同步操作 VS	适合异步操作，但可能引入额外开销
     * <p><br/>
     * 为了让 flatMap 的异步操作在指定的线程池中执行，可以使用以下操作符：
     * <li>publishOn(Scheduler)将后续操作切换到指定的调度器。
     * <li>subscribeOn(Scheduler)将整个数据流的订阅过程切换到指定的调度器。
     * <p><br/>
     * 根据任务类型选择合适的调度器
     * <li>Schedulers.immediate()	在当前线程执行（默认行为）
     * <li>Schedulers.single()	在单个专用线程执行
     * <li>Schedulers.parallel()	CPU 密集型任务
     * <li>Schedulers.boundedElastic()	I/O 密集型任务（如网络请求、数据库查询）
     */
    @Test
    public void testFlatMap() {
        // flatMap: 默认行为（在当前线程执行）Schedulers.immediate()
        Mono.just("Hello")
                .flatMap(s -> Mono.fromCallable(() -> {
                    System.out.println("Processing on thread: " + Thread.currentThread().getName());
                    return s.toUpperCase();
                }))
                .subscribe(System.out::println);

        // 使用 publishOn 切换到新的线程池
        Mono.just("Hello")
                .publishOn(Schedulers.parallel()) // 切换到并行线程池
                .flatMap(s -> Mono.fromCallable(() -> {
                    System.out.println("Processing on thread: " + Thread.currentThread().getName());
                    return s.toUpperCase();
                })).subscribe(System.out::println);
        // 使用 subscribeOn 切换到新的线程池
        Mono.just("Hello")
                .flatMap(s -> Mono.fromCallable(() -> {
                    System.out.println("Processing on thread: " + Thread.currentThread().getName());
                    return s.toUpperCase();
                }))
                .subscribeOn(Schedulers.boundedElastic()) // 切换到弹性线程池
                .subscribe(System.out::println);
    }

    /**
     * zip 操作符将多个数据流中的元素成对组合，形成一个新的项，每次从每个流中取出一个元素，直到任何一个流中的元素用尽。
     * 它按照指定的顺序组合元素，确保每个组合都是成对的。
     * zip 输出的数量是最短输入流的长度。
     */
    @Test
    public void testZip() {
        Flux<String> flux1 = Flux.just("A", "B", "C");
        Flux<Integer> flux2 = Flux.range(1, 3);

        // 使用 zip 将两个流合并为一个新的流，每个元素是一个包含原始流的对应元素的元组
        Flux<String> zippedFlux = Flux.zip(flux1, flux2, (s, i) -> s + "-" + i);

        zippedFlux.subscribe(m -> logger.info("received: {}", m)); // 输出可能是 A-1, B-2, C-3
    }

    /**
     * <li>特性：	Mono.just()	；Mono.fromCallable()	；Mono.defer()
     * <li>执行时机：	声明时立即执行	；订阅时执行	；订阅时执行
     * <li>适用场景：	静态值	；无副作用的同步逻辑	；有副作用或动态生成的逻辑
     * <li>返回值：	固定的值	；通过 Callable 生成的值	；通过 Supplier 生成的 Mono
     * <p><br/>
     * <li>如果需要在订阅时动态生成 Mono，或者创建逻辑有副作用，使用 Mono.defer()。
     * <li>如果只需要将同步的 Callable 转换为 Mono，使用 Mono.fromCallable()。
     * <li>如果值是静态的，使用 Mono.just()。
     */
    @Test
    public void testDefer() {
        // 使用 Mono.defer() 避免副作用
        Mono<String> deferMono = Mono.defer(() -> {
            System.out.println("Creating Mono..."); // 副作用：打印日志
            return Mono.just("Hello");
        });

        // 声明时不会执行副作用
        System.out.println("Mono 声明");

        // 订阅时执行副作用
        deferMono.subscribe(System.out::println);
    }

    /**
     * 注意事项
     * <li>doOnTerminate 方法中的操作是在 Mono 终止时执行的，无论终止是正常完成还是出现错误。
     * <li>如果 Mono 被取消，doOnTerminate 中的操作不会被执行。
     */
    @Test
    public void testDoOnTerminate() {
        System.out.println("-----------------------------正常执行");
        Mono.just("正常结束")
                .doOnTerminate(() -> logger.info("Mono doOnTerminate!"))
                .subscribe(
                        value -> logger.info("Received: {}", value),
                        error -> logger.info("Error: {}", error.getMessage()),
                        () -> logger.info("Completed")
                );
        System.out.println("-----------------------------异常执行");
        Mono.error(new RuntimeException("异常执行"))
                .doOnTerminate(() -> logger.info("doOnTerminate!"))
                .doOnError(error -> logger.info("doOnError: {}", error.getMessage()))
                .subscribe(
                        value -> logger.info("Received: {}", value),
                        error -> logger.info("errorConsumer : {}", error.getMessage()),
                        () -> logger.info("Completed")
                );
        System.out.println("-----------------------------取消执行");
        Disposable disposable = Mono.delay(Duration.ofSeconds(2)) // 延迟 2 秒
                .map(v -> "Data")
                .doOnTerminate(() -> logger.info("doOnTerminate!!取消后，doOnTerminate不再执行"))
                .doOnCancel(() -> logger.info("doOnCancel"))
                .subscribe(
                        value -> System.out.println("Received: " + value),
                        error -> System.err.println("Error: " + error),
                        () -> System.out.println("Completed")
                );
        // 主线程等待 1 秒后取消订阅
        ThreadDemo.safeSleep(1000);
        logger.info("执行取消操作");
        disposable.dispose(); // 取消订阅

        // 主线程等待以确保程序不会立即退出
        ThreadDemo.safeSleep(3000);
    }


    @Test
    public void testDoOnCancel() {
        System.out.println("-----------------------------使用 timeout 自动取消");
        // 创建一个延迟 3 秒的 Mono
        Mono<String> mono = Mono.delay(Duration.ofSeconds(3))
                .map(value -> "Data")
                .timeout(Duration.ofSeconds(2)) // 设置超时时间为 2 秒
                .doOnCancel(() -> logger.info("doOnCancel: doOnCancel 的执行条件是 订阅被显式取消，而 timeout 操作符的行为是 触发超时错误，而不是取消订阅。"));

        // 订阅
        mono.subscribe(
                value -> logger.info("consumer Received: {}", value),
                error -> logger.info("errorConsumer: {}", error),
                () -> logger.info("completeConsumer : Completed")
        );

        // 主线程等待以确保程序不会立即退出
        ThreadDemo.safeSleep(4000);

    }


}

class Resource implements AutoCloseable {
    public Resource() {
        System.out.println("创建：Resource created");
    }

    public String getData() {
        System.out.println("读取：Reading data");
        return "Some data";
    }

    @Override
    public void close() {
        System.out.println("关闭：Resource closed");
    }
}
